package com.sheepone.distributor;

import com.lmax.disruptor.EventTranslator;
import com.sheepone.distributor.event.DistributeEvent;
import com.sheepone.distributor.event.EventContext;
import com.sheepone.distributor.listener.DistributorAwareListener;
import com.sheepone.distributor.listener.DistributorListener;
import com.sheepone.distributor.listener.DistributorListenerHolder;
import com.sheepone.util.algorithm.analysis.IntegerAnalysisUtil;
import com.sheepone.util.algorithm.math.RandomUtil;
import com.sheepone.util.number.NumberUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import static com.sheepone.distributor.listener.DistributorListenerMasker.*;

/**
 * @author Miss.杨
 * @description
 * @since 2023-12-13 20:11
 */
public abstract class Distributor {
    protected static final Logger logger = LoggerFactory.getLogger(Distributor.class);
    protected static final AtomicInteger CREATED = new AtomicInteger(0);

    // todo 待探究：Disruptor.halt()之后还可以start()吗
    protected final AtomicBoolean stopped = new AtomicBoolean(true);
    protected final String name;
    protected final List<DistributorLine> lines;

    protected int oSize;

    protected final List<DistributorListenerHolder> holders = new ArrayList<>();


    public Distributor(String name, int size, List<DistributorLine> lines) {
        if (!CREATED.compareAndSet(0, 1) && logger.isWarnEnabled()) {
            CREATED.incrementAndGet();
            logger.warn("It is not advocated for multiple Distributors to run simultaneously. " +
                    "If you want to expand the Distributors, " +
                    "you can use the resize() method");
        }
        this.name = name;
        this.oSize = size;
        this.lines = lines;
    }

    public String getName() {
        return name;
    }

    public void distribute(String targetStation, String targetTopic, EventContext context) {
        if (stopped.get()) {
            throw new RuntimeException("Distributor has been stopped");
        }
        lines.get(RandomUtil.randomIndex(lines.size()))
                .publish((e, s) -> {
                            e.setContext(context);
                            e.setTargetStation(targetStation);
                            e.setTargetTopic(targetTopic);
                        }
                );
    }

    protected void distribute(EventTranslator<DistributeEvent> translator) {
        if (stopped.get()) {
            throw new RuntimeException("Distributor has been stopped");
        }
        lines.get(RandomUtil.randomIndex(lines.size())).publish(translator);
    }

    public abstract void distributeDelay(String targetStation, String targetTopic, EventContext context, long targetMillisStamp);

    public Distributor start() {
        if (stopped.get() && stopped.compareAndSet(true, false)) {
            beforeStarting();
            lines.forEach(DistributorLine::start);
            logger.info("【Distributor】 \"{}\" start to work", name);
            afterStarting();
        }
        return this;
    }

    public void stop() {
        if (!stopped.get() && stopped.compareAndSet(false, true)) {
            beforeStopping();
            lines.forEach(DistributorLine::stop);
            WorkerStationManager.INSTANCE.stopAll();
            logger.info("【Distributor】 \"{}\" has stopped", name);
            afterStopping();
        }
    }

    public void resize(int size) {
        if (size < 1) {
            throw new IllegalArgumentException("size must be greater than 0");
        }
        if (!stopped.get()) {
            int tsize = IntegerAnalysisUtil.upper2Pow(size / lines.size());
            if (size != IntegerAnalysisUtil.upper2Pow(oSize / lines.size())) {
                lines.forEach(line -> line.resize(tsize));
            }
            oSize = size;
        }
    }

    public long hasPublishNum() {
        return lines.stream().mapToLong(DistributorLine::hasPublishedNum).sum();
    }

    public void addListeners(DistributorListener... listeners) {
        if (stopped.get()) {
            for (DistributorListener listener : listeners) {
                if (listener instanceof DistributorAwareListener awareListener) {
                    awareListener.setDistributor(this);
                }
                this.holders.add(new DistributorListenerHolder(listener));
            }
        }
    }

    private List<DistributorListener> executedListeners(int mask) {
        return holders.stream()
                .filter(holder -> holder.isMarked(mask))
                .map(DistributorListenerHolder::getListener)
                .toList();
    }

    private void beforeStarting() {
        executedListeners(MASK_BEFORE_STARTING).forEach(DistributorListener::beforeStarting);
    }

    private void afterStarting() {
        executedListeners(MASK_AFTER_STARTING).forEach(DistributorListener::afterStarting);
    }

    private void beforeStopping() {
        executedListeners(MASK_BEFORE_STOPPING).forEach(DistributorListener::beforeStopping);
    }

    private void afterStopping() {
        executedListeners(MASK_AFTER_STOPPING).forEach(DistributorListener::afterStopping);
    }

    @Override
    public String toString() {
        return "Distributor{" +
                "name='" + name + '\'' +
                '}';
    }
}
